
package cc.jinglupeng.kom.pool;

import cc.jinglupeng.kom.bean.ZkBroker;
import kafka.javaapi.consumer.SimpleConsumer;
import org.apache.commons.pool.KeyedPoolableObjectFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

public class KafkaConsumerFactory implements KeyedPoolableObjectFactory<Integer, SimpleConsumer> {

    private Map<Integer, ZkBroker> zkBrokers;
    private boolean fetchBrokerInfoReady = false;
    private Logger logger = LoggerFactory.getLogger(KafkaConsumerFactory.class);
    private int bufferSize = 64 * 1024;
    private int timeout = 3 * 1000;
    private String clientId = "KOM";


    public KafkaConsumerFactory(Map<Integer, ZkBroker> zkBrokers) {
        this.zkBrokers = zkBrokers;
    }


    @Override
    public SimpleConsumer makeObject(Integer key) throws Exception {
        Map<Integer, ZkBroker> zkBrokers = getZkBrokers();
        if (zkBrokers == null) {
            throw new Exception("brokers cannot be null");
        }
        ZkBroker zkBroker = zkBrokers.get(key);
        if (zkBroker == null) {
            throw new Exception("cannot get broker info for key " + key);
        }
        return new SimpleConsumer(zkBroker.getHost(), zkBroker.getPort(), getTimeout(), getBufferSize(), "KOM");
    }

    @Override
    public void destroyObject(Integer integer, SimpleConsumer simpleConsumer) throws Exception {
        if (simpleConsumer != null) {
            simpleConsumer.close();
        }
    }

    @Override
    public boolean validateObject(Integer integer, SimpleConsumer simpleConsumer) {
        return true;
    }

    @Override
    public void activateObject(Integer integer, SimpleConsumer simpleConsumer) throws Exception {
    }

    @Override
    public void passivateObject(Integer integer, SimpleConsumer simpleConsumer) throws Exception {
    }

    public int getBufferSize() {
        return bufferSize;
    }

    public void setBufferSize(int bufferSize) {
        this.bufferSize = bufferSize;
    }

    public String getClientId() {
        return clientId;
    }

    public void setClientId(String clientId) {
        this.clientId = clientId;
    }

    public Map<Integer, ZkBroker> getZkBrokers() {
        return zkBrokers;
    }

    public void setZkBrokers(Map<Integer, ZkBroker> zkBrokers) {
        this.zkBrokers = zkBrokers;
    }

    public int getTimeout() {
        return timeout;
    }

    public void setTimeout(int timeout) {
        this.timeout = timeout;
    }
}
